package com.nageoffer.jobtrain.common.framework.idempotent;

import com.nageoffer.jobtrain.common.framework.enums.IdempotentMQConsumeStatusEnum;
import com.nageoffer.jobtrain.common.framework.toolkit.SpELUtil;
import com.nageoffer.jobtrain.common.web.exception.ServiceException;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.aspectj.lang.ProceedingJoinPoint;
import org.aspectj.lang.annotation.Around;
import org.aspectj.lang.annotation.Aspect;
import org.aspectj.lang.reflect.MethodSignature;
import org.springframework.data.redis.core.StringRedisTemplate;
import org.springframework.data.redis.core.script.RedisScript;
import org.springframework.stereotype.Component;

import java.lang.reflect.Method;
import java.util.List;
import java.util.Objects;
import java.util.concurrent.TimeUnit;

/**
 * 防止消息队列消费者重复消费消息切面控制器
 */
@Aspect
@Slf4j
@RequiredArgsConstructor
@Component
public final class NoMQDuplicateConsumeAspect {
    private final StringRedisTemplate stringRedisTemplate;

    /**
     * ('SET',KEY,value,'NX','GET','PX',expire_time_ms)
     * 'SET',KEY,value：在Redis中设置 key对应的value值
     * 'NX'：不存在才进行设置
     * 'GET'：设置新值之前，返回之前的旧值
     * 'PX',expire_time_ms：设置key的过期时间为 expire_time_ms
     */
    private static final String LUA_SCRIPT = """
            local key = KEYS[1]
            local value = ARGV[1]
            local expire_time_ms = ARGV[2]
            return redis.call('SET', key, value, 'NX', 'GET', 'PX', expire_time_ms)
            """;

    /**
     * 增强方法标记 {@link NoMQDuplicateConsume} 注解执行逻辑
     */
    @Around("@annotation(com.nageoffer.jobtrain.common.framework.idempotent.NoMQDuplicateConsume)")
    public Object noMQDuplicateConsume(ProceedingJoinPoint joinPoint) throws Throwable {
        // 先获取到这个防幂等性注解
        NoMQDuplicateConsume noMQDuplicateConsume = getNoMQDuplicateConsume(joinPoint);
        // 通过 SpEL 解析机制，完成动态解析，并得到幂等性缓存 Redis
        String uniqueKey = noMQDuplicateConsume.keyPrefix() + SpELUtil.parseKey(noMQDuplicateConsume.key(),
                ((MethodSignature) joinPoint.getSignature()).getMethod(), joinPoint.getArgs());
        /**
         * 这里会有三种可能
         * absentAndGet：当第一次通过Lua脚本进行设置状态时为执行中，返回 null
         * absentAndGet：当有相同的消息过来时，当前消息还在处理中，直接抛出异常，返回之前状态 0
         * absentAndGet：当再次有相同消息过来时，此时已经消费完全，不作处理，返回 1
         */
        String absentAndGet = stringRedisTemplate.execute(
                RedisScript.of(LUA_SCRIPT, String.class),
                List.of(uniqueKey),
                IdempotentMQConsumeStatusEnum.CONSUMING.getCode(),
                String.valueOf(TimeUnit.SECONDS.toMillis(noMQDuplicateConsume.keyTimeout()))
        );
        // 如果不为空，证明已经有
        if (Objects.nonNull(absentAndGet)) {
            boolean errorFlag = IdempotentMQConsumeStatusEnum.isError(absentAndGet);
            log.warn("[{}] MQ重复消费，{}。", uniqueKey, errorFlag ? "等待客户端延迟消费" : "状态已完成");
            // 此时消费状态为消费中，抛出异常
            if (errorFlag) {
                throw new ServiceException(String.format("消息消费者幂等异常，幂等标识：%s", uniqueKey));
            }
            return null;
        }

        Object result;
        try {
            // 执行原方法逻辑
            result = joinPoint.proceed();
            // 设置防重复消费令牌 Key过期时间，单位秒，状态执行完成
            stringRedisTemplate.opsForValue().set(uniqueKey, IdempotentMQConsumeStatusEnum.CONSUMED.getCode(),
                    noMQDuplicateConsume.keyTimeout(), TimeUnit.SECONDS);
        } catch (Throwable ex) {
            // 执行失败，直接删除幂等Key，让消息再次进行消费
            stringRedisTemplate.delete(uniqueKey);
            throw ex;
        }
        return result;
    }

    /**
     * @return 返回防止消息队列重复消费消息幂等性注解
     */
    private static NoMQDuplicateConsume getNoMQDuplicateConsume(ProceedingJoinPoint joinPoint) throws NoSuchMethodException {
        MethodSignature methodSignature = (MethodSignature) joinPoint.getSignature();
        Method targetMethod = joinPoint.getTarget().getClass()
                .getDeclaredMethod(methodSignature.getName(), methodSignature.getMethod().getParameterTypes());
        return targetMethod.getAnnotation(NoMQDuplicateConsume.class);
    }
}